package aggregator;

import akka.actor.ActorRef;
import com.espertech.esper.client.*;
import controllers.HistoryController;
import controllers.MessagesController;
import controllers.ObjectsController;
import entities.Objects;
import helpers.DistanceCalculator;
import helpers.LocationCalculator;
import helpers.LocationHelper;
import helpers.MessageActor;
import play.Logger;
import play.libs.Akka;
import play.mvc.Controller;
import scala.concurrent.duration.Duration;

import javax.persistence.PostPersist;
import java.io.FileNotFoundException;
import java.io.PrintWriter;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;

import static java.util.concurrent.TimeUnit.SECONDS;

public class EsperAggregator implements Runnable {

    public static EPServiceProvider engine;
    public static ConcurrentHashMap<Integer/*tagNumber*/,LocationHelper> mLocHelper= new ConcurrentHashMap<Integer,LocationHelper>();
    public static ConcurrentHashMap<String, HashMap<String, Double>> currentEvents = new ConcurrentHashMap<String, HashMap<String, Double>>();
    public static PrintWriter pWriter;
    public static long buttonClickTime;
    public static void main(String[] args) {
        new EsperAggregator().run();
    }




	public Integer distance;
	public boolean buttonClicked = false;

    public EsperAggregator(){
        //The Configuration is meant only as an initialization-time object.
        Configuration cepConfig = new Configuration();
        // We register Ticks as objects the engine will have to handle
        //cepConfig.addEventType("StockTick",<span class="skimlinks-unlinked">Tick.class.getName</span>());

        distance = 0;
        // We setup the engine
        engine = EPServiceProviderManager.getProvider("InTracks",cepConfig);
        try {
			pWriter = new PrintWriter("check.txt");
		} catch (FileNotFoundException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
//	    ConfigurationHTTPAdapter adapterConfig = new ConfigurationHTTPAdapter();
//
//	    Service service = new Service();
//	    service.setPort(8079);
//	    
//	    adapterConfig.getServices().put("inTracks", service);
//	    
//	    GetHandler get = new GetHandler();
//	    get.setPattern("*");
//	    get.setService("inTracks");
//	    
//	    adapterConfig.getGetHandlers().add(get);
//		
//		// start adapter
//		EsperIOHTTPAdapter httpAdapter = new EsperIOHTTPAdapter(adapterConfig, "InTracks");
//		httpAdapter.start();

        
        EPStatement createTagSighting = engine.getEPAdministrator().createEPL("create schema TagSighting as (tag String, reader String, power int,button int)");
        EPStatement selectAvgTagSighting = engine.getEPAdministrator().createEPL("select tag, reader, avg(power) from TagSighting.win:time(10 sec) group by tag, reader");
        EPStatement selectButtonTagSighting = engine.getEPAdministrator().createEPL("select tag, button from TagSighting(button != 0)");

        TagSightingListener sightingListener = new TagSightingListener();
        selectAvgTagSighting.addListener(sightingListener);

        ButtonClickListener buttonClickListener = new ButtonClickListener();
        selectButtonTagSighting.addListener(buttonClickListener);


        EPStatement createTagLocation = engine.getEPAdministrator().createEPL("create schema TagLocation as (tag String, reader String, x int,y int)");
        EPStatement createStations = engine.getEPAdministrator().createEPL("create schema Stations as (id long, x int, y int, name String)");

        //create a namedWindow
        try{
        	EPStatement createStation = EsperAggregator.engine.getEPAdministrator().createEPL("create window StationsTable.win:keepAll() as select * from Stations");
        }catch(Exception e){
        	try{
        	//EPStatement createStation = EsperAggregator.engine.getEPAdministrator().createEPL("delete from StationsTable");
        	}catch(Exception e1){
        		
        	}
        	//EPStatement createStation = EsperAggregator.engine.getEPAdministrator().createEPL("@Drop select * from Stations");
    }        
       
        //select tag location
        EPStatement selectTagLocation = engine.getEPAdministrator().createEPL("select * from TagLocation");
        TagLocationListener listenerTagLocation = new TagLocationListener();
        selectTagLocation.addListener(listenerTagLocation);
        

        //Tag Insert to station - send a new message (notification) and update location helper
          EPStatement tagOutFromStation = engine.getEPAdministrator().createEPL("select null as id, sta.name as name ,tag.tag as tag"
          																	+  " from TagLocation.win:time(30 sec) as tag"
          																	+  " left outer join StationsTable.std:unique(id) as sta "
          																	+  " where controllers.TagsController.isInStation(sta.id,sta.x,sta.y,tag.x,tag.y) is null");
          TagOutFromStationListener listenerTagOutFromStation = new TagOutFromStationListener();
          tagOutFromStation.addListener(listenerTagOutFromStation);
          
          
        //Tag Insert to station - send a new message (notification) and update location helper
        EPStatement tagInsertToStation = engine.getEPAdministrator().createEPL("select sta.id as id, sta.name as name ,tag.tag as tag"
        																	+  " from TagLocation.win:time(30 sec) as tag"
        																	+  " left outer join StationsTable.std:unique(id) as sta "
        																	+  " where controllers.TagsController.isInStation(sta.id,sta.x,sta.y,tag.x,tag.y) is not null");
        TagInsertToStationListener listenerTagInsertToStation = new TagInsertToStationListener();
        tagInsertToStation.addListener(listenerTagInsertToStation);
        
        
        //Tag Insert to station - update location helper
        EPStatement selectUpdateHistory = engine.getEPAdministrator().createEPL("select controllers.TagsController.isInStation(sta.id,sta.x,sta.y,tag.x,tag.y) as id ,tag.tag as tag,tag.x as x, tag.y as y "
        																	+  " from TagLocation.win:time(30 sec) as tag"
        																	+  " left outer join StationsTable.std:unique(id) as sta ");
        UpdateHistoryListener listenerUpdateHistory = new UpdateHistoryListener();
        selectUpdateHistory.addListener(listenerUpdateHistory);
        
        //select Stations and insert new row to namedWindow StationTable
        EPStatement selectStations = engine.getEPAdministrator().createEPL("select * from Stations");
        StationsListener listenerstation = new StationsListener();
        selectStations.addListener(listenerstation);

    }

    public static List<CurrentEvents> getCurrentEvents(){
        ArrayList<CurrentEvents> currentEventses = new ArrayList<CurrentEvents>();
        CurrentEvents currentEvent;
        for(Object obj : currentEvents.keySet()){
            currentEvent = new CurrentEvents();
            currentEvent.setTagCode(obj.toString());
            List<String> readers = new ArrayList<String>();
            for(Object reader : currentEvents.get(obj).keySet()){
                readers.add(reader+" power: "+currentEvents.get(obj).get(reader));
            }
            
            currentEvent.setReaders(readers);
            currentEventses.add(currentEvent);
        }


        return currentEventses;
    }

    /*ITAY*/
    public class UpdateHistoryListener implements UpdateListener {
        public void update(EventBean[] newEvents, EventBean[] oldEvents) {
        	for (EventBean eventBean : newEvents) {
        		Integer tagNumber = new Integer((String)eventBean.get("tag"));
        		if(tagNumber != null){
        			Long stationId = null;
        			if(eventBean.get("id") != null){
        				stationId = (long)eventBean.get("id");
        			}
        			int locationX = (int)eventBean.get("x");
        			int locationY = (int)eventBean.get("y");
        			try {
						HistoryController.insertOrUpdateHistory(stationId, tagNumber, locationX, locationY);
					} catch (Throwable e) {
						// TODO Auto-generated catch block
						e.printStackTrace();
					}
        		}
			
			}
        }
    }
    
    /*ITAY*/
    public class StationsListener implements UpdateListener {
        public void update(EventBean[] newEvents, EventBean[] oldEvents) {
        	EPStatement insertStation = engine.getEPAdministrator().createEPL("insert into StationsTable(id, x, y, name) select id, x , y ,name from Stations");

        }
    }
    
    /*ITAY*/
    public class TagLocationListener extends Controller implements UpdateListener {
  	  public void update(EventBean[] newEvents, EventBean[] oldEvents) {

          for (EventBean tag : newEvents)
          {
          	Integer tagNumber = new Integer((String)tag.get("tag"));
          	LocationHelper locHelper;
              if (tagNumber != null)
              {
              	locHelper = mLocHelper.get(tagNumber);
              	if(locHelper != null){
              		locHelper.setLocationX(new Integer((int)tag.get("x")));
              		locHelper.setLocationY(new Integer((int)tag.get("y")));
              	}else{
                  	locHelper = new LocationHelper();
                  	locHelper.setTagNumber(tagNumber);
                  	try {
						locHelper.setObjects(ObjectsController.getObjectByTag(tagNumber));
					}catch(ClassCastException c){
						
					}
                  	catch (Throwable e) {
						// TODO Auto-generated catch block
						e.printStackTrace();
					}
                  	locHelper.setLocationX(new Integer((int)tag.get("x")));
              		locHelper.setLocationY(new Integer((int)tag.get("y")));
              		mLocHelper.put(tagNumber, locHelper);
                  }
              }
              
          }
      }
  }
    
    /*ITAY*/
    public class TagInsertToStationListener implements UpdateListener {
    	@PostPersist
        public void update(EventBean[] newEvents, EventBean[] oldEvents) {
        	 
        	for (EventBean tag : newEvents)
            {
            	Integer tagNumber = new Integer((String)tag.get("tag"));
            	Long stationId = (Long)tag.get("id");
            	String stationName = (String)tag.get("name");

            	if(stationId != null){
	            	Objects o = null;
					try {
						o = ObjectsController.getObjectByTag(tagNumber);
					}catch(ClassCastException e){
						
					}
					catch (Throwable e) {
						e.printStackTrace();
					}
					LocationHelper loc = mLocHelper.get(tagNumber);
					loc.setObjects(o);
					
					//if the object is exist
					if(o != null){
						
	            		if(loc.getStationName() == null || !loc.getStationName().equals(stationName) ){
	            			loc.setStationId(stationId);
	            			loc.setStationName(stationName);
	            			
			            	String message = o.getName() + " got in the " + stationName;

			            	MessagesController.saveMessage(message,Boolean.TRUE);
			            	
			            	Akka.system().scheduler().scheduleOnce(
			                        Duration.create(0, SECONDS),
			                        MessageActor.defaultMessage,
			                        MessagesController.saveMessage(message, true),
			                        Akka.system().dispatcher(),
			                        ActorRef.noSender()
			                );
	            		}
	            	}
	            }
            }
        }
    }
    
    /*ITAY*/
    public class TagOutFromStationListener implements UpdateListener {
    	@PostPersist
        public void update(EventBean[] newEvents, EventBean[] oldEvents) {
        	 
    		int i = 0;
        	for (EventBean tag : newEvents)
            {
        		i++;
            	Integer tagNumber = new Integer((String)tag.get("tag"));
            	Long stationId = (Long)tag.get("id");
            	String stationName = (String)tag.get("name");

            	if(stationId == null){
	            	Objects o = null;
					try {
						o = ObjectsController.getObjectByTag(tagNumber);
					}catch(ClassCastException e){
						
					}
					catch (Throwable e) {
						e.printStackTrace();
					}
					LocationHelper loc = mLocHelper.get(tagNumber);
					loc.setObjects(o);
					
					//if the object is exist
					if(o != null){
						//station id
						
	            		if(loc.getStationName() != null && loc.getStationName().equals(stationName)){
	            			
	            			loc.setStationId(null);
	            			loc.setStationName(null);
	            			
			            	String message = o.getName() + " out from last station ";
			            	
			            	MessagesController.saveMessage(message,Boolean.TRUE);
			            	
			            	Akka.system().scheduler().scheduleOnce(
			                        Duration.create(0, SECONDS),
			                        MessageActor.defaultMessage,
			                        MessagesController.saveMessage(message, true),
			                        Akka.system().dispatcher(),
			                        ActorRef.noSender()
			                );
	            		}
	            	}
	            }
            }
        }
    }
    
    public class TagSightingListener implements UpdateListener {
        public void update(EventBean[] newEvents, EventBean[] oldEvents) {
            HashMap<String, Double> curReaders;
        /*    Logger.info("*********************** Tags Sighted (last 3 sec): ********************");
*/
            for (EventBean tag : newEvents)
            {
            	Double power = (Double) tag.get("avg(power)");
            	 
            	if(buttonClicked  && System.currentTimeMillis() < buttonClickTime + (10 * 1000) && power != null )
            	{
    				pWriter.println(power+ ","+ distance);
    			 	Logger.info("***************************************["+power+ ","+ distance+"]***************************************");
    			 	pWriter.flush();
            	}
            	else if (System.currentTimeMillis() >= buttonClickTime + (10 * 1000))
            		buttonClicked = false;
            	 
                if (tag.get("avg(power)") != null) {
                    if(currentEvents.containsKey(tag.get("tag"))){
                        currentEvents.get(tag.get("tag")).put(tag.get("reader").toString(), (Double)tag.get("avg(power)"));
                       
                    }else{
                        curReaders = new HashMap<String, Double>();
                        curReaders.put(tag.get("reader").toString(), (Double)tag.get("avg(power)"));
                        currentEvents.put(tag.get("tag").toString(), curReaders);
                        
                    }

                    //Calculate distance to reader based on average signal strength in the last 10 seconds
                    DistanceCalculator distCalc = DistanceCalculator.getCalcs().get(tag.get("reader").toString());

                    Float distance = distCalc.calcDistance(((Double)tag.get("avg(power)")).floatValue());

                    // Add the current distance to the location estimator, which outputs a new location for every 3 reader seen in the last second
                    LocationCalculator.getCalc(tag.get("tag").toString()).addDistance(tag.get("reader").toString(), distance, distCalc.getX(), distCalc.getY());
                    
//                    Logger.info("tag=" + tag.get("tag") + "\treader=" + tag.get("reader") + "\tpower=" + tag.get("avg(power)"));
                }
                else
                	Logger.info("tag=" + tag.get("tag") + " not seen in the last 10 sec.");
            }
        }
    }
    
    
    public class ButtonClickListener implements UpdateListener {
        public void update(EventBean[] newEvents, EventBean[] oldEvents) {

         /*   Logger.info("*********************** Tags Sighted (last 3 sec)  tid:"+ Thread.currentThread().getId() +": ********************");
*/
            for (EventBean tag : newEvents)
            {
            	if(tag.get("button") != null && !tag.get("button").equals(new Integer(0)))
        		{ 
            		if(System.currentTimeMillis() >= buttonClickTime + (10 * 1000))
        			{
        				 distance++;
        				 buttonClickTime = System.currentTimeMillis();
        				 buttonClicked = true;
        			}  
        		}
             //   if (tag.get("x") != null)
                  //  Logger.info("tid:"+Thread.currentThread().getId() +" tag=" + tag.get("tag") + "\treader=" + tag.get("reader") + "\tX=" + tag.get("x")+ "\tY=" + tag.get("y"));
             //   else
                 //   Logger.info("tag=" + tag.get("tag") + " not seen in the last 10 sec.");

            }

         /*   Logger.info("***********************************************************************");*/

        }
    }

   


    @Override
    public void run() {
        // TODO Auto-generated method stub



        while (true){
            try {
                Thread.sleep(100);
            } catch (InterruptedException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
        }

        // destroy the adapter when done
        //httpAdapter.destroy();
    }

    public static class CurrentEvents{
        String tagCode;
        List<String> readers;

        public String getTagCode() {
            return tagCode;
        }

        public void setTagCode(String tagCode) {
            this.tagCode = tagCode;
        }

        public List<String> getReaders() {
            return readers;
        }

        public void setReaders(List<String> readers) {
            this.readers = readers;
        }
    }
}
